package com.sbp.message.schedule;

import com.alibaba.fastjson.JSON;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.sbp.message.api.entity.Message;
import com.sbp.message.api.entity.MessageStatus;
import com.sbp.message.api.service.MessageStatusAckService;
import com.sbp.message.common.MyDubboReferenceContainer;
import com.sbp.message.send_message.MessageSender;
import com.sbp.message.service.cs.MessageCSImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;

public class UnconfirmedMessageCheckSchedule implements SimpleJob {
    private final static Logger logger = LoggerFactory.getLogger(UnconfirmedMessageCheckSchedule.class);

    @Autowired
    private MessageCSImpl messageCSImpl;
    @Autowired
    private MessageSender messageSender;
    @Autowired
    private MyDubboReferenceContainer dubboReferenceContainer;


    public void execute(ShardingContext shardingContext) {
        int shardingItem = shardingContext.getShardingItem();
        int shardingTotalCount = shardingContext.getShardingTotalCount();
        logger.debug("状态确认任务, 当前时间戳={}", System.currentTimeMillis());

        // 获取状态为待确认的消息
        List<Message> messageList = messageCSImpl.selectShardingListByStatusAndElapseTime(MessageStatus.unconfirmed, 30, shardingItem, shardingTotalCount);

        List<String> trashMessageIdList = new ArrayList<>();

        for (Message message : messageList) {
            try {
                MessageStatusAckService messageStatusAckService = dubboReferenceContainer.getReference(message.getSource() + ".messageStatus.ack", MessageStatusAckService.class);

                if (messageStatusAckService.acknowledge(message.getId())) {
                    logger.info("will confirm and send unconfirmed message={}", JSON.toJSONString(message));
                    message.setStatus(MessageStatus.sending.getValue());
                    messageCSImpl.updateById(message);
                    messageSender.sendMessage(message);
                } else {
                    // 半个小时还没有确认才会删除消息
                    if (System.currentTimeMillis() - message.getCreateTime().getTime() >= 1800 * 1000) {
                        trashMessageIdList.add(message.getId());
                        logger.info("will remove trash message={}", JSON.toJSONString(message));
                    }
                }
            } catch (Throwable throwable) {
                logger.info("", throwable);
            }
        }

        // 删除垃圾数据
        if (!trashMessageIdList.isEmpty()) {
            messageCSImpl.batchDelete(trashMessageIdList);
        }
    }

}
